% Created 2021-02-01 Mon 16:20
% Intended LaTeX compiler: pdflatex
\documentclass[11pt]{article}
\usepackage[utf8]{inputenc}
\usepackage[T1]{fontenc}
\usepackage{graphicx}
\usepackage{grffile}
\usepackage{longtable}
\usepackage{wrapfig}
\usepackage{rotating}
\usepackage[normalem]{ulem}
\usepackage{amsmath}
\usepackage{textcomp}
\usepackage{amssymb}
\usepackage{capt-of}
\usepackage{hyperref}
\usepackage{ctex}
\author{lin chuan}
\date{\today}
\title{}
\hypersetup{
 pdfauthor={lin chuan},
 pdftitle={},
 pdfkeywords={},
 pdfsubject={},
 pdfcreator={Emacs 26.3 (Org mode 9.3.8)}, 
 pdflang={English}}
\begin{document}

\tableofcontents

\section{Chapter 2. Thread Safety}
\label{sec:orgeb31457}
\subsection{Locking}
\label{sec:org857e0b5}

A Synchronized block has tow parts:

\begin{itemize}
\item a reference to an object that will serve as the lock

\item a block of code to be guarded by that lock
\end{itemize}


A \texttt{synchronized} method is a shorthand for a \texttt{synchronized} block that spans an entire method body, and whose lock is th object on which the method is being invoked.

Stack \texttt{synchronized} method use the \texttt{Class} object for the lock.

\begin{verbatim}
synchronized (lock) {
    // Access or modify shared state guarded by lock
}
\end{verbatim}

每个Java对象都可以作为一个以同步为目的的锁, 这些内置锁为称为 \texttt{intrinsic locks} 或是 \texttt{monitor locks}. 

Instrinsic locks又被称之为互斥锁, 意味着最多只有一个线程可以拥有这个锁.

Java的 \textasciitilde{}intrinsic lock\textasciitilde{}是可重入锁, 如果一个线程可以请求它已经获得的锁.

重入为每个锁分配一个计数器和所占有的线程.


这个行为和POSIX线程的互斥锁机制不一样, POSIX的互斥锁是基于调用的, 而Java的互斥锁是基于线程的.

Java的重入锁机制简化了面向对象的开发流程, 因为如果不是重入锁, 下面的代码就会出现死锁:

\begin{verbatim}
public class Widget {
    public synchronized void doSomething() {
	// ...
    }
}


public class LoggingWidget extends Widget {
    public synchronized void doSomething() {
	System.out.println(toString() + ": calling doSomething");
	super.doSomething();
    }
}
\end{verbatim}

\subsection{Guarding State with Locks}
\label{sec:org4ce9ad3}

For each mutable state variable that may be accessed by more than one thread, all accesses to that variable must be performed with the same lock held. In this case, we say that the variable is guarded by that lock.

对于可能被超过一个线程访问的可变状态变量, 所有对于这个变量的访问都需要加上相同的锁. 在这个场景下, 我们说这个变量被这个锁保护.

对象的内部锁和对象的状态之间并无内在的联系, 我们仅仅是为了简化构建锁对象的流程, 从而使得每个对象都有一个内部锁. 注意: 为一个对象加锁并没有阻止其他线程访问该对象, 为对象加锁的实际意图是为了阻止其他线程获取这个锁, 重点的是锁.

每个共享的, 可变的变量最好都被不同的锁所保护. 写代码时明确这一点可能更有利于维护.

并非所有的数据都需要加锁, 那些会被多个线程访问的可变数据才需要加锁.

如果一个Class包含的不变式由多个变量组成, 则需要为这几个变量的访问分配相同的锁.

啥叫不变式: 符合一个特定的规则, 例如各家银行的总钱数在借贷过程中保持不变.

\subsection{Liveness and Performance}
\label{sec:org1751cbd}

一个例子:

\begin{verbatim}
public class CachedFactorizer extends HttpServlet {

  private BigInteger lastNumber;
  private BigInteger[] lastFactors;

  private long hits;
  private long cacheHits;

  @Override
  public void service(ServletRequest servletRequest, ServletResponse servletResponse)
      throws ServletException, IOException {
    BigInteger i = new BigInteger(servletRequest.getParameter("i"));
    BigInteger[] factors = null;

    synchronized (this) {
      ++hits;
      if (i.equals(lastNumber)) {
	factors = lastFactors.clone();
	cacheHits++;
      }
    }

    if (factors == null) {
      factors = factor(i);
      synchronized (this) {
	lastNumber = i;
	lastFactors = factors.clone();
      }
    }

    servletResponse.getWriter().append(Arrays.toString(factors));
  }

  private BigInteger[] factor(BigInteger i) {
    var r = new BigInteger[i.intValue()];
    r[0] = r[1] = BigInteger.valueOf(1);
    for (int j = 2; j < r.length; j++) {
      r[j] = r[j - 1].add(r[j - 2]);
    }
    return r;
  }
}
\end{verbatim}

\section{Chapter 3. Sharing Objects}
\label{sec:org823739b}
\subsection{3.1 Visibility}
\label{sec:org71f1c70}

可见性

\begin{verbatim}
public class NoVisibility {
  private static boolean ready;
  private static int number;

  private static class ReaderThread implements Runnable {
    @Override
    public void run() {
      while (!ready) Thread.yield();
      System.out.println(number);
    }
  }

  public static void main(String[] args) {
    new ReaderThread().run();
    number = 42;
    ready = true;
  }
}
\end{verbatim}

上面代码的运行结果完全不可预测, 因为在为同步的情况下, 两个线程对于共享数据的修改和访问取决与编译器及JVM指令.

有可能reader线程永远无法读取到ready为true的值, 也可能是先读取到了true, 然后读取到的number值为0.

\subsubsection{3.1.1 Stale Data}
\label{sec:org4e71592}

\subsubsection{3.1.2 Nonatomic 64-bit Operations}
\label{sec:org20097af}

据说 \texttt{double} 和 \texttt{long} 没有被申明为volatile.

\subsubsection{3.1.3 Locking and Visibility}
\label{sec:orgac09bbc}

加锁不仅是为了互斥, 还是为了内存的可见性.

\url{./img/ch03fig01.gif}

\subsubsection{3.1.4 Volatile Variables}
\label{sec:orgf0977b2}

When a field is declared volatile, the compiler and runtime are put on notice that this variable is shared and that operations on it should not be reordered with other memory operations. Volatile variables are not cached in registers or in caches where they are hidden from other processors, so a read of a volatile variable always returns the most recent write by any thread.

当一个字段被定义为 \texttt{volatile}, 编译器和运行环境会认为这个变量是共享变量, 对该变量的操作不应该和其他内存操作乱序. \texttt{Volatile} 变量也不会在寄存器和cache中缓存, 因此对一个 \texttt{volatile} 变量的读操作总会返回该变量的最近的值.

Use volatile variables only when they simplify implementing and verifying your synchronization policy; avoid using volatile variables when veryfing correctness would require subtle reasoning about visibility. Good uses of volatile variables include ensuring the visibility of their own state, that of the object they refer to, or indicating that an important lifecycle event (such as initialization or shutdown) has occurred.

一个 \texttt{Volatile} 变量的示例:

\begin{verbatim}
volatile boolean asleep;

while (!asleep)
    countSomeSheep();
\end{verbatim}

在上面这个示例中, 使用 \texttt{volatile} 变量可以确保每次都能读取到 \texttt{asleep} 的最新状态, 而 \texttt{volatile} 通常也是用在这个场景下:

\begin{itemize}
\item 某个变量会被多个线程写, 那么通过将该变量定义为 \texttt{volatile} , 可以确保每次读取到最新的状态.
\end{itemize}


Locking can guarantee both visibility and atomicity; volatile variables can only guarantee visibility.

加锁可以确保可见性和原子性, 而 \texttt{volatile} 变量仅仅能保证可见性.

You can use volatile variables only when all the following criteria are met:

\begin{itemize}
\item Writes to the variable do not depend on its current value, or you can ensure that only a single thread ever updates the value;

\item The variable does not participate in invariants with other state variables; and

\item Locking is not required for any other reason while the variable is being accessed.
\end{itemize}

\subsection{3.2 Publication and Escape}
\label{sec:org1fd1f01}

最直白的发布是在public static字段中存放引用:

\begin{verbatim}
public static Set<Secret> knownSecrets;

public void initialize() {
    knownSecrets = new HashSet<Secret>();
}
\end{verbatim}

在上面的示例中, 如果你向 \texttt{knownSecrets} 中添加了一个 \texttt{Secret}, 那么这个 \texttt{Secret} 也被发布出去了.

类似的, 返回一个引用的非私有方法也会将这个引用所指向的对象发布出去:

\begin{verbatim}
class UnsafeStates {
    private String[] states = new String[] {
	"AK", "AL" //...
    };
    public String[] getStates() {return states;}
}
\end{verbatim}

在上面的示例中, 虽然看似用的get方法, 然后返回的是对象的引用, 那么就可以修改这个对象. 因此, 我们说 \texttt{states} 数组从它的作用域中逃离了.

不好的示例:

\begin{verbatim}
public class ThisEscape {
    public ThisEscape(EventSource source) {
	source.registerListener(
		new EventListener() {
		    public void onEvent(Event e) {
			doSomething(e);
		    }
		});
    }
}
\end{verbatim}

因为在构造函数中启用了新的进程, 从而导致this对象可以在新的线程中被使用, 而新的进程开启时, 当前的this对象可能还没有初始化完成.

\subsubsection{3.2.1 Safe Construction Practices}
\label{sec:org0922fd2}

But an object is in a predictable, consistent state only after its constructor returns, so publishing an object from within its constructor can publish an incompletely constructed object . If the this reference escapes during construction, the object is considered not properly constructed.

一个对象只有当其构造函数返回后才是完整状态, 因此在构造函数中发布这个对象可能会导致该对象被使用的时候不完整 (构造不完全)

More specifically, the this reference should not escape from the thread until after the constructor returns. The this reference can be stored somewhere by the constructor so long as it is not used by another thread until after construction.

应该先将this引用存放起来, 等到构造完成后在被其他线程使用.

\begin{verbatim}
public class SafeListener {
    private final EventListener listener;

    private SafeListener() {
	listener = new EventListener() {
		public void onEvent(Event e) {
		    doSomething(e);
		}
	    };
    }

    public static SafeListener newInstance(EventSource source) {
	SafeListener safe = new SafeListener();
	source.registerListener(safe.listener);
	return safe;
    }
}
\end{verbatim}

上述示例中, 1. 构造函数是私有访问, 2. 只能通过 public static方法来获取新的实例, 3. 确保了新的线程启动的时候拿到的SafeListener对象是完整的.

\subsection{3.3 Thread Confinement}
\label{sec:orgf776380}

如果数据仅能被一个线程所访问, 那么这个数据天然就是线程安全的. 例如JDBC连接池中的Connection对象, 连接池可以被多个线程访问, 进程获取连接池中的一个connection, 使用它来处理单个请求, 使用完后归还到连接池中. 连接池有一个单独的线程负责connection的分配, 如果这个connection未归还, 则不会将它分配给其他线程, 以此来确保connection对象的线程安全.

\subsubsection{3.3.1. Ad-hoc Thread Confinement}
\label{sec:org12cf08f}

特殊/临时的线程绑定, 靠代码来实现, 例如只有一个线程可以修改某个变量的值, 而其他变量都是读取, 这个变量定义为 \texttt{volatile}, 以确保可见性.

备用, 不太经常使用, 因为要靠代码来保证.

\subsubsection{3.3.2. Stack Confinement}
\label{sec:org2979ef2}

在线程栈空间中的局部变量都是栈绑定的, 只有当前栈才能访问到这个局部变量.

栈绑定又被成为 \texttt{within-thread} 或是 \texttt{thread-loca}, 注意和下面的 \texttt{ThreadLocal} 区分开来.

关键的一点, 不要让这些局部变量逃离他们的作用域, 例如返回一个引用之类的.

\subsubsection{3.3.3. ThreadLocal}
\label{sec:org4d61fe6}

\begin{verbatim}
private static ThreadLocal<Connection> connectionHolder
    = new ThreadLocal<Connection>() {
	    public Connection initialValue() {
		return DriverManager.getConnection(DB_URL);
	    }
	};

public static Connection getConnection() {
    return connectionHolder.get();
}
\end{verbatim}

When a thread calls \texttt{ThreadLocal.get} for the first time, \texttt{initialValue} is consulted to provide the initial value for that thread. Conceptually, you can think of a \texttt{ThreadLocal<T>} as holding a \texttt{Map<Thread,T>} that stores the thread-specific values, though this is not how it is actually implemented. The thread-specific values are stored in the Thread object itself; when the thread terminates, the thread-specific values can be garbage collected.

\subsection{3.4. Immutability}
\label{sec:orga86470c}

An object is immutable if:

\begin{itemize}
\item Its state cannot be modifled after construction;
\item All its flelds are final;[12] and

[12] It is technically possible to have an immutable object without all fields being final—String is such a class—but this relies on delicate reasoning about benign data races that requires a deep understanding of the Java Memory Model. (For the curious: String lazily computes the hash code the first time hashCode is called and caches it in a nonfinal field, but this works only because that field can take on only one nondefault value that is the same every time it is computed because it is derived deterministically from immutable state. Don't try this at home.)

\item It is properly constructed (the this reference does not escape during construction).
\end{itemize}

这一节在讲不变对象, 注意final修饰的变量有可能是一个引用, 那么需要认真考虑该引用指向的对象是否不可变.

\begin{verbatim}
public final class ThreeStooges {
    private final Set<String> stooges = new HashSet<String>();

    public ThreeStooges() {
	stooges.add("Moe");
	stooges.add("Larry");
	stooges.add("Curly");
    }

    public boolean isStooge(String name) {
	return stooges.contains(name);
    }
}
\end{verbatim}

\subsubsection{3.4.1. Final Fields}
\label{sec:org51b7cee}

Final fields can't be modified (although the objects they refer to can be modified if they are mutable), but they also have special semantics under the Java Memory Model. It is the use of final fields that makes possible the guarantee of initialization safety (see Section 3.5.2) that lets immutable objects be freely accessed and shared without synchronization.

\texttt{final class} 指的是这个类不能被继承

\subsubsection{3.4.2. 示例 - 使用Volatile发布不可变对象}
\label{sec:org4ce803b}

\begin{verbatim}
class OneValueCache {
  private final BigInteger lastNumber;
  private final BigInteger[] factors;

  public OneValueCache(BigInteger lastNumber, BigInteger[] factors) {
    this.lastNumber = lastNumber;
    this.factors = factors.clone();
  }

  public BigInteger[] getFactors(BigInteger i) {
    if (i != null && i.equals(lastNumber)) {
      return factors.clone();
    } else {
      return null;
    }
  }
}

public class VolatileCachedFactorizer implements Servlet {

  private volatile OneValueCache cache = new OneValueCache(null, null);

  @Override
  public void service(ServletRequest servletRequest, ServletResponse servletResponse)
      throws ServletException, IOException {

    BigInteger i = new BigInteger(servletRequest.getParameter("index"));

    var factors = cache.getFactors(i);
    if (factors == null) {
      factors = factor(i);
      cache = new OneValueCache(i, factors);
    }

    servletResponse.getWriter().append(Arrays.toString(factors));
  }

  private BigInteger[] factor(BigInteger i) {
    var r = new BigInteger[i.intValue()];
    r[0] = r[1] = BigInteger.valueOf(1);
    for (int j = 2; j < r.length; j++) {
      r[j] = r[j - 1].add(r[j - 2]);
    }
    return r;
  }
}
\end{verbatim}

\subsection{3.5. Safe Publication}
\label{sec:org1219c50}

\begin{verbatim}
// Unsafe publication
public Holder holer;

public void initialize() {
    holder = new Holder(42);
}
\end{verbatim}

上面的示例为什么不是安全的发布呢??? 我也没看懂

You may be surprised at how badly this harmless-looking example could fail. Because of visibility problems, the Holder could appear to another thread to be in an inconsistent state, even though its invariants were properly established by its constructor! This improper publication could allow another thread to observe a partially constructed object.

\subsubsection{3.5.1. Imporoper Publication: When Good Objects Go Bad}
\label{sec:orge5dde19}

\begin{verbatim}
public class Holder {
    private int n;

    public Holder(int n) { this.n = n; }

    public void assertSanity() {
	if (n != n)
	    throw new AssertionError("This statement is false.");
    }
}
\end{verbatim}

Because synchronization was not used to make the Holder visible to other threads, we say the Holder was not properly published.

\subsubsection{3.5.2. Immutable Objects and Initialization Safety}
\label{sec:org717c658}

For this guarantee of initialization safety to hold, all of the requirements for immutability must be met: unmodi-fiable state, all fields are final, and proper construction.

\begin{itemize}
\item 状态不可变

\item 所有的字段都是final

\item 合适的构造函数
\end{itemize}

\subsubsection{3.5.3. Safe Publication Idioms}
\label{sec:orgeddec71}

To publish an object safely, both the reference to the object and the object's state must be made visible to other threads at the same time. A properly constructed object can be safely published by:

\begin{itemize}
\item Initializing an object reference from a static initializer;

\item Storing a reference to it into a volatile field or AtomicReference;

\item Storing a reference to it into a final field of a properly constructed object; or

\item Storing a reference to it into a field that is properly guarded by a lock.
\end{itemize}


为了做到安全的发布一个对象, 对象的引用以及对象的状态对其他线程应该同时可见. 合适的构造函数应该满足以下条件:

\begin{itemize}
\item 从静态初始化器(static initializer) 初始化对象的引用

\item 将对象的引用保存在 \texttt{volatile} 或是 \texttt{AtomicReference} 的字段中

\item 将对象的引用存放在 \texttt{final} 字段

\item 将对象的引用存放在被锁保护的字段
\end{itemize}

The thread-safe library collections offer the following safe publication guarantees, even if the Javadoc is less than clear on the subject:

\begin{itemize}
\item Placing a key or value in a Hashtable, synchronizedMap, or Concurrent-Map safely publishes it to any thread that retrieves it from the Map (whether directly or via an iterator);
\item Placing an element in a Vector, CopyOnWriteArrayList, CopyOnWrite-ArraySet, synchronizedList, or synchronizedSet safely publishes it to any thread that retrieves it from the collection;
\item Placing an element on a BlockingQueue or a ConcurrentLinkedQueue safely publishes it to any thread that retrieves it from the queue.
\end{itemize}


Using a static initializer is often the easiest and safest way to publish objects that can be statically constructed:

\texttt{public static Holder holder = new Holder(42);}

\subsubsection{3.5.4. Effectively Immutable Objects}
\label{sec:orgd5e477d}

这个好像不是在说特性, 是说有些对象本身是可变的, 但程序中并不会改变它们的值, 因此也可以把这些对象当作是不变对象.

For example, Date is mutable,[17] but if you use it as if it were immutable, you may be able to eliminate the locking that would otherwise be required when sharing a Date across threads. Suppose you want to maintain a Map storing the last login time of each user:

\subsubsection{3.5.5. Mutable Objects}
\label{sec:orge79afca}

The publication requirements for an object depend on its mutability:

\begin{itemize}
\item Immutable objects can be published through any mechanism;
\item Effectively immutable objects must be safely published;
\item Mutable objects must be safely published, and must be either \textbf{threadsafe} or \textbf{guarded by a lock}.
\end{itemize}

\subsubsection{3.5.6. Sharing Objects Safely}
\label{sec:orgf09d8c1}

Whenever you acquire a reference to an object, you should know what you are allowed to do with it. Do you need to acquire a lock before using it? Are you allowed to modify its state, or only to read it? Many concurrency errors stem from failing to understand these “rules of engagement” for a shared object. When you publish an object, you should document how the object can be accessed.

获取一个对象的引用时, 需要考虑:
\begin{itemize}
\item 是否需要获得锁
\item 是否可以修改它的状态
\item 是否是只读
\end{itemize}

发布一个对象时, 需要说明这个 \textbf{对象应该如何访问}.

The most useful policies for using and sharing objects in a concurrent program are:

\begin{itemize}
\item \textbf{Thread-confined}. A thread-confined object is owned exclusively by and confined to one thread, and can be modifled by its owning thread.

\item \textbf{Shared read-only}. A shared read-only object can be accessed concurrently by multiple threads without additional synchronization, but cannot be modified by any thread. Shared read-only objects include immutable and effectively immutable objects.

\item \textbf{Shared thread-safe}. A thread-safe object performs synchronization internally, so multiple threads can freely access it through its public interface without further synchronization.
\end{itemize}

\section{Chapter 4. Composing Objects}
\label{sec:orgfeaa888}

如何构造线程安全的类

\subsection{4.1. Designing a Thread-Safe Class}
\label{sec:org29f9b40}

The design process for a thread-safe class should include these three basic elements:

\begin{itemize}
\item Identify the variables that form the object's state;标识构成对象状态的变量
\item Identify the invariants that constrain the state variables;表示状态变量的不变式
\item Establish a policy for managing concurrent access to the object's state.建立管理并发访问的策略
\end{itemize}

\subsubsection{4.1.1. Gathering Synchronization Requirements}
\label{sec:orgf25d6e9}

构造线程安全的类意味着确保对象的不变式在并行访问时也能保持. 因此需要关注对象的状态. 对象和变量都有"状态空间": 可能的状态范围. 

注意在尽可能的情况下使用 \texttt{final} 成员变量, 可以更好的控制对象的可能状态.

操作可能有 \texttt{postconditions} 用来标识某些不合法的状态迁移.

由不变式和 \texttt{postconditions} 约束的状态迁移是我们创建同步或封装需求的来源.

如果 \textbf{某些状态是非法的, 那么底层的状态变量必须进行封装}, 否则用户代码可能会将对象置为非法状态. \textbf{如果一个操作可能导致非法的状态变迁, 那么它必须是原子的}

Multivariable invariants like this one create atomicity requirements: related variables must be fetched or updated in a single atomic operation. You cannot update one, release and reacquire the lock, and then update the others, since this could involve leaving the object in an invalid state when the lock was released. When multiple variables participate in an invariant, the lock that guards them must be held for the duration of any operation that accesses the related variables.

如果一个不变式由多个变量组成, 那么对于这些变量的读写操作必须在单独的原子操作中完成, 不能更新一个的值, 释放并获取锁, 再更新其他变量的值.

You cannot ensure thread safety without understanding an object's \textbf{invariants} and \textbf{postconditions}. Constraints on the valid values or state transitions for state variables can create atomicity and encapsulation requirements.

不变式和 \texttt{postconditions} 是原子和封装的需求来源!!!

\subsubsection{4.1.2. State-dependent Operations}
\label{sec:orgd8d7e1a}

Some objects also have methods with state-based preconditions. For example, you cannot remove an item from an empty queue; a queue must be in the “nonempty” state before you can remove an element. Operations with state-based preconditions are called state-dependent

一些对象的方法可能由基于状态的前置条件, 例如, 你不能在一个空队列中移除元素. 对于这些基于状态的前置条件的操作称之为 \textbf{状态依赖}

在单线程环境中, 如果前置条件不满足, 那么操作只能失败; 而在多线程环境下, 如果前置条件当前时间不满足, 则可以等待其状态变化. \textbf{增加了等待的可能性}

内置锁(Instrinsic Lock)有wait和notify方法可以调用

其他的阻塞库(Blocking Library)有 \textbf{BlockingQueue}, \textbf{Semaphore} 可以使用

\subsubsection{4.1.3. State Ownership}
\label{sec:org8c22e63}

当定义组成对象状态的变量时, 我们仅考虑那些对象所拥有的数据.

If you allocate and populate a HashMap, you are creating multiple objects: the HashMap object, a number of Map.Entry objects used by the implementation of HashMap, and perhaps other internal objects as well. The logical state of a HashMap includes the state of all its Map.Entry and internal objects, even though they are implemented as separate objects.

举 \texttt{HashMap} 的例子来看, 当你分配并发布一个 \texttt{HashMap} 时, 你创建了多个对象:
\texttt{HashMap} 对象本身, 一些 \texttt{Map.Entry} 对象, 还有可能其他的内部对象. 一个 \texttt{HashMap} 对象状态应该包含其 \texttt{Map.Entry} 的状态, 尽管它们可能是由不同的对象所构成.

\subsection{4.2.Instance Confinement}
\label{sec:orgc67f85a}

实例封闭

通过实例封闭, 可以使得更为简单的使用封装来实现线程安全.

当一个对象被封装到另一个对象之中, 这个被封装的对象更可以清楚的被分析出有那些路径可以被访问到, 其分析过程比从整个程序中来访问这个对象要更为直观.

在一个实例封闭中的数据只能通过外层对象的方法进行访问, 使得更为简单的确保所有对于里层数据的访问都被合适的锁保护.

\subsubsection{4.2.1. The Java Monitor Pattern}
\label{sec:org741d306}

An object following the Java monitor pattern encapsulates all its mutable state and guards it with the object's own intrinsic lock.

Java \textbf{monitor 模式}: 将可变状态都封装起来, 并通过对象的内在锁进行保护

Java的 \texttt{intrinsic lock} 有时被称之为 \texttt{monitor lock} 或 \texttt{monitors}

\begin{verbatim}
public final class Counter {
    private long value = 0;

    public synchronized long getValue() {
	return value;
    }

    public synchronized long increment() {
	if (value == Long.MAX_VALUE)
	    throw new IllegalStateException("counter overflow");
	return ++value;
    }
}
\end{verbatim}

任意的锁对象都可以用来保持对象的状态:

\begin{verbatim}
public class PrivateLock {
    private final Object myLock = new Object();
    Widget Widget;

    void someMethod() {
	synchronized(myLock) {
	    // Access or modify the state of widget
	}
    }
}
\end{verbatim}

使用私有对象作为锁还有一个好处: \textbf{这个私有成员对象无法被用户调用}, 确保了代码的安全性.

\subsubsection{4.2.2. Example: Tracking Fleet Vehicles}
\label{sec:orgf5bd7d5}

\subsection{4.3 Delegating Thread Safety}
\label{sec:org00a057e}

\subsubsection{4.3.1. Example: Vehicle Tracker Using Delegating}
\label{sec:org838ce92}

\subsubsection{4.3.2. Independent State Variables}
\label{sec:orga9f7b21}

\begin{verbatim}
public class VisualComponent {
  private final List<KeyListener> keyListeners = new CopyOnWriteArrayList<>();
  private final List<MouseListener> mouseListeners = new CopyOnWriteArrayList<>();

  public void addKeyListener(KeyListener keyListener) {
    keyListeners.add(keyListener);
  }

  public void addMouseListener(MouseListener mouseListener) {
    mouseListeners.add(mouseListener);
  }

  public void removeKeyListener(KeyListener keyListener) {
    keyListeners.remove(keyListener);
  }

  public void removeMouseListener(MouseListener mouseListener) {
    mouseListeners.remove(mouseListener);
  }
}
\end{verbatim}

\texttt{VisualComponent} 使用 \texttt{CopyOnWriteArrayList} 来存放 \texttt{listener}.

\texttt{CopyOnWriteArrayList} 是线程安全的, 因此可以说 \texttt{VisualComponent} 将其对象的线程安全委托给了其两个线程安全的成员变量.

\subsubsection{4.3.3. When Delegation Fails}
\label{sec:org4939936}

一个委托失败的例子:

\begin{verbatim}
public class NumberRange {
  // INVARIANT:lower <= upper
  private final AtomicInteger lower = new AtomicInteger(0);
  private final AtomicInteger upper = new AtomicInteger(0);

  public void setLower(int i) {
    if (i > upper.get()) throw new IllegalArgumentException("Can't set lower to " + i + " > upper");
    lower.set(i);
  }

  public void setUpper(int i) {
    if (i < lower.get()) throw new IllegalArgumentException("Can't set upper to " + i + " < lower");
    upper.set(i);
  }

  public boolean isInRange(int i) {
    return (i >= lower.get() && i <= upper.get());
  }
}
\end{verbatim}

\texttt{NumberRange} 不是线程安全的, 因为并没有满足不变式: \texttt{lower <= upper}.

可以通过使用锁来保护它的不变式, 同时 \texttt{lower} 和 \texttt{upper} 不能发布出去, 以免用户更改了其状态, 造成不变式不成立.

如果一个类由多个独立的线程安全的状态变量组成, 而且没有非法的状态变迁操作, 那么这个类可以将线程安全委托给其底层的状态变量.

\subsubsection{4.3.4. 发布底层状态变量}
\label{sec:org80d6a82}

一般来说, 不要将状态变量暴露出去, 仅在某些情况下可以这样做:

\begin{itemize}
\item 该状态变量本身是线程安全的, 例如原子类
\item 该类的不变式不包含这个状态变量
\item 对该变量的任意操作都不会造成非法的状态变迁
\end{itemize}

以上三个条件必须同时满足, 例如之前的 \texttt{VisualComponent} 中的两个listeners就满足这几个条件, 即使暴露出去也不会造成现场安全的问题.

\subsubsection{4.3.5. Example: Vehicle Tracker that Publishes Its State}
\label{sec:org9b10393}

\subsection{4.4 Adding Functionality to Existing Thread-Safe Classes}
\label{sec:org25a1cd3}

\textbf{在已有的线程安全类上增加功能}

\begin{itemize}
\item 继承, 并在新功能上加锁

\item 使用Java monitor模式, 将原有类的对象作为新类的成员变量
\end{itemize}

\subsubsection{Client-side Locking}
\label{sec:orgfef6966}

失败的示例:

\begin{verbatim}
public class ListHelper<E> {
    public List<E> list = Collection.synchronizedList(new ArrayList<E>());

    //...

    public synchronized boolean pubIfAbsent(E x) {
	boolean absent = !list.contains(x);
	if (absent)
	    list.add(x);
	return absent;
    }
}
\end{verbatim}

上面问题的原因在于加的锁并不是list的锁, \texttt{putIfAbsent} 对于list的操作和list的其他操作并不是互斥的, 因此并非线程安全. 

这种情况下需要去查看list的锁是哪一个, 用对应的锁来保证线程安全:

\begin{verbatim}
public class ListHelper<E> {
    public List<E> list = Collection.synchronizedList(new ArrayList<E>());

    //...

    public boolean pubIfAbsent(E x) {
	synchronized(list) {
	    boolean absent = !list.contains(x);
	    if (absent)
		list.add(x);
	    return absent;
	}
    }
}
\end{verbatim}

\subsubsection{4.4.2. Composition}
\label{sec:orga75f64f}

\begin{verbatim}
public class ImprovedList<T> implements List<T> {
  private final List<T> list;

  public ImprovedList(List<T> list) {
    this.list = list;
  }

  public synchronized boolean putIfAbsent(T x) {
    boolean contains = list.contains(x);
    if (!contains) list.add(x);
    return !contains;
  }

  public synchronized void clear() {
    list.clear();
  }
}
\end{verbatim}

不管底层的list是否是线程安全的, 使用 \texttt{ImporvedList} 自己的内置锁来保证线程安全.

\subsection{4.5. Documenting Synchronization Policies}
\label{sec:org48b8adc}

\section{Chapter 5. Building Blocks}
\label{sec:org1cda29f}

介绍常用的并发构建方法, 以及使用他们来构造并发应用的设计模式.

\subsection{5.1. Synchronized Collections}
\label{sec:orgbd1ddcb}

同步的集合类包括 \texttt{Vector}, \texttt{HashTable}, 以及由 \texttt{Collections.synchronizedXxx} 工厂方法创建的同步包装类.

\subsubsection{5.1.1. 同步集合的问题}
\label{sec:org416fe59}

同步集合整体上来说是线程安全的, 但是有时需要增加额外的client-side锁来确保复合操作的线程安全. 常用的复合操作包括:

\begin{itemize}
\item iteration(迭代, 重复拿出元素直至集合耗尽)
\item navigation(导航, 按照某种顺序找到当前元素的下一个元素)
\item condition operations(条件操作, put-if-absent)
\end{itemize}

这些符合操作需要注意其线程安全性, 例如下面的操作就不是线程安全的

\begin{verbatim}
public static Object getLast(Vector list) {
    int lastIndex = list.size() - 1;
    return list.get(lastIndex);
}

public static void deleteLast(Vector list) {
    int lastIndex = list.size() - 1;
    list.remove(lastIndex);
}
\end{verbatim}

\url{./img/ch05fig01.gif}

所以上述两个操作, 需要去查看Vector使用什么作为锁, 一边来说集合类都是使用的其对象自身的内置锁:

\begin{verbatim}
public static Object getLast(Vector list) {
    synchronized(list) {
	int lastIndex = list.size() - 1;
	return list.get(lastIndex);
    }
}

public static void deleteLast(Vector list) {
    synchronized(list) {
	int lastIndex = list.size() - 1;
	list.remove(lastIndex);
    }
}
\end{verbatim}

同样的, 集合类元素的遍历也会有类似的问题:

\begin{verbatim}
for (int i = 0; i < vector.size(); i++)
    doSomething(vector.get(i));
\end{verbatim}

这段代码中获取第i个元素的get方法有可能会失败(被其他线程删除了第i个元素)

\begin{verbatim}
synchronized(vector) {
    for (int i = 0; i < vector.size(); i++)
	doSomething(vector.get(i));
}
\end{verbatim}

虽然通过上面加锁的方法可以保证遍历时的线程安全, 但是却阻止了其他线程对于数据的访问, 伤害了并发性.

\subsubsection{5.1.2. Iterators and Concurrentmodificationexception}
\label{sec:org9609eb5}

迭代 \texttt{Collection} 的标准方法是使用迭代器(\textbf{\texttt{Iterator}})

然而使用Iterator也并没有减少迭代时加锁的需求, 由安全集合返回的迭代器并没有设计为处理并发访问, 它们是 \texttt{fail-fast} 的, 意思就是检测到迭代期间集合发生了变化就抛出 \texttt{ConcurrentModificationException} 异常. 一般是在 \texttt{hasNext} 或 \texttt{next} 方法中抛出. 通过关联一个修改计数器到集合上来探测异常, 如果修改计数器在迭代的过程中发生了改变, 则抛出异常. 尽管如此, 这个检查并没有定义为同步的, 因此也有看见修改计数器脏值的可能, 导致线程不安全.

\begin{verbatim}
List<widget> widgetList = Collection.synchronizedList(new ArrayList<Widget>());
// ...

// May throw ConcurrentModificationException
for (Widget w: widgetList)
    doSomething(w);
\end{verbatim}

在迭代过程中并未给整个集合加锁是有原因的: 

\begin{itemize}
\item 其他的线程可能需要访问集合的数据

如果集合较大, 或是每个元素的处理时间较长, 那么其他线程的等待时间就更久

\item 死锁的风险

上述代码中, \texttt{doSomethin()} 在锁中被调用, 有可能会造成死锁(Chapter 10)

\item 影响性能

即使没有死锁和饿死, 对一个资源加锁时间过长, 导致等待这个资源的线程变多, 锁消耗的CPU资源就更多
\end{itemize}

与加锁不同, 保证迭代线程安全的另一个方案是克隆并遍历克隆, 因为克隆后的对象是栈对象, Thread-Confinement, 栈封闭的, 因此不存在线程安全问题. 需要注意的是克隆操作也需要加锁. 克隆操作也是有相当开销的, 需要衡量开销的代价.

\subsubsection{5.1.3. Hidden Iterators}
\label{sec:org03a68ab}

集合类的 \texttt{toString}, \texttt{equals}, \texttt{hashCode} 都隐式调用了迭代器的迭代方法

使用集合类作为参数的 \texttt{containsAll}, \texttt{removeAll}, \texttt{retainAll} 以及构造函数, 也都隐式的调用了迭代器及迭代方法.

所有这些非直接使用也都可能导致 \texttt{ConcurrentModificationException} 异常.

\subsection{5.2. Concurrent Collections}
\label{sec:orgef11da2}

上一节的同步集合类虽然保证了线程安全, 但是并发性比较糟糕, 因此出现了一些支持高并发的集合类:

\begin{itemize}
\item ConcurrentHashMap (替代 \texttt{HashTable} )
\item CopyOnWriteArrayList (主要操作是遍历时, 效率较高)
\item ConcurrentMap (提供了诸如 \texttt{pub-if-absent}, \texttt{replace}, \texttt{conditional remove} 的符合操作)
\end{itemize}

\subsubsection{5.2.1. ConcurrentHashMap}
\label{sec:org5134151}

\texttt{ConcurrentHashMap} 不会抛出 \texttt{ConcurrentModificationException}, 由 \texttt{ConcurrentHashMap} 返回的迭代器被称之为 \texttt{weakly consistent} 而不是 \texttt{fail-fast}. \texttt{ConcurrentHashMap} 在遍历的时候并未对整个对象加锁, 从而允许多个线程同时读写

A weakly consistent iterator can tolerate concurrent modification, traverses elements as they existed when the iterator was constructed, and may (but is not guaranteed to) reflect modifications to the collection after the construction of the iterator.

一些可能的问题: \texttt{size} 以及 \texttt{isEmpty} 未必准确, 因为这两个操作获取的是当前状态的结果, 并未加锁处理. 然后在并发环境中, 更重要的操作也许是诸如 \texttt{get}, \texttt{put}, \texttt{containsKey}, \texttt{remove} 之类.

\texttt{HashMap} , \texttt{SynchronizedMap} 与 \texttt{ConcurrentHashMap} 相比的另一个差异点是: 这两个类可以对元素做互斥限制, 禁止其他线程的访问, 在某些场景下有用: 添加多个映射??, 按照一定顺序遍历多次.

\subsubsection{5.2.2. Additional Atomic Map Operations}
\label{sec:org748ba66}

\texttt{ConcurrentHashMap} 没有互斥操作的锁, 因此无法使用 \texttt{client-side} 锁来创造新的原子操作(诸如 put-if-absent之类); 如果需要 \texttt{pub-if-absent}, \texttt{remove-if-equal}, \texttt{replace-if-equal} 这样的符合操作, 需要考虑使用 \texttt{ConcurrentMap} 接口

这段话不理解, 上面这些方法虽然是在 \texttt{ConcurrentMap} 中定义的, 但是在 \texttt{ConcurrentHashMap} 中已经实现了这些方法\ldots{}\ldots{}

\subsubsection{5.2.3. CopyOnWriteArrayList}
\label{sec:orgd13b9af}

\begin{verbatim}
public void add(int index, E element) {
    synchronized (lock) {
	Object[] es = getArray();
	int len = es.length;
	if (index > len || index < 0)
	    throw new IndexOutOfBoundsException(outOfBounds(index, len));
	Object[] newElements;
	int numMoved = len - index;
	if (numMoved == 0)
	    newElements = Arrays.copyOf(es, len + 1);
	else {
	    newElements = new Object[len + 1];
	    System.arraycopy(es, 0, newElements, 0, index);
	    System.arraycopy(es, index, newElements, index + 1,
			     numMoved);
	}
	newElements[index] = element;
	setArray(newElements);
    }
}
\end{verbatim}

读取的时候没有锁, 修改的时候直接copy一份原来的数据, 这样其他线程还可以访问到之前的数据, 提升了并发效能.

copy操作会有较大的性能开销, 因此 \texttt{CopyOnWriteArrayList} 主要适用于遍历比修改多得多的场景.

\subsection{5.3. Blocking Queues and The Producer-Consumer Pattern}
\label{sec:org6e2c788}

多个阻塞队列的实现包括: \texttt{LindedBlockingQueue}, \texttt{ArrayBlockingQueue}, 都是FIFO队列, 另为还有 \texttt{PriorityBlockingQueue} 是优先级队列.

最后一个 \texttt{BlockingQueue} 的实现是 \texttt{SynchronousQueue}, 并非是真的队列, 因为没有排队元素的存储空间, 它维护排队现场的列表(等待将元素入队列或出队列的线程). \texttt{SynchronousQueue} 因为没有中间的队列, 所以减小了数据搬移的时间, 而且可以返回更多的任务状态给生产者(因为它直接和消费者通信).更适用于有足够多的消费者处理任务的场景.

\subsubsection{5.3.1. Example: Desktop Search}
\label{sec:orgf3ecf75}

\subsubsection{5.3.2. Serial Thread Confinement (串行化的线程约束)}
\label{sec:orgd27b4eb}

可变对象从生产者交付给消费者后, 生产者就不要再操作该对象, 否则会导致线程不安全.

\subsubsection{5.3.3. Deques and Work Stealing}
\label{sec:org54d7f0a}

Java 6 引入了两个新的集合类型: \texttt{Deque} (读做 "deck") 和 \texttt{BlockingDeque}. 具体实现包括 \texttt{ArrayDeque} 和 \texttt{LinkedBlockingDeque}.

Deques可以用于 \texttt{work stealing} 场景: 每个工人都有自己的deque, 但是如果他的工作做完了, 就可以从其他工人的deque去拿工作过来做. \^{}\_\^{}

做自己的工作从队列头开始拿, 做别人的工作从队列尾拿, 这样可以减小竞争, 提升并发性能.

另外, 这个模式适用于自己给自己找活干的场景: 即是生产者, 又是消费者. 例如网页爬虫: 分析一个网页的内容, 可能会产生更多的网页代分析, 同样的内容, 那么新的工作被加入到deque的队尾.

\subsection{{\bfseries\sffamily DONE} 5.4. Blocking and Interruptible Methods}
\label{sec:org9c0457e}
线程可能被阻塞, 暂停: 等待I/O完成, 等待锁, 等待从 \texttt{Thread.sleep} 中唤醒, 等待其他线程的计算结果\ldots{}当一个现场被阻塞, 它就进入了阻塞状态(BLOCKED, WAITING, TIMED\textsubscript{WATITING}), 从而等待某个事件的发生, 当外部事件发生变化时, 这个线程重新被标记为 \texttt{RUNNABLE}, 参与线程调度的竞争.

\texttt{BlockingQueue} 中的 \texttt{put}, \texttt{take} 方法会抛出 checked \texttt{InterruptedException}. 这种会抛出 \texttt{InterruptedException} 异常的方法都被称为阻塞方法, 当其被中断(interrupted)时, 会尽力结束阻塞状态.

\texttt{Thread} 提供了 \texttt{interrupt} 方法来终端线程, (或是询问线程是否已经被中断 \texttt{Thread.currentThread().isInterrupted()} ). 每个线程都有一个boolean属性用来表示其中断状态.

Interruption是互相配合的一个机制, 一个线程不能强制另一个线程停止现在的工作. 阻塞方法就是相应中断请求的地方, 能够取消长时间运行的事务.

当你的代码中调用了阻塞方法, 那么应该处理中断请求 - 即处理 \texttt{InterruptedException} 异常. 这主要有两个选择:

\begin{enumerate}
\item 向上传递异常, 在最外层去处理

\item 设置中断状态

有时你没法继续向上抛出异常, 例如你在实现了 \texttt{Runnable} 接口. 在这种情况下, 你必须捕获异常, 并设置 \texttt{interruped status} (通过在当前线程中调用 \texttt{Thread.currentThread().interrupt()} 方法), 以便上层调用栈可以看到异常发生. 外面一层的代码可以通过调用 \texttt{Thread.currentThread().isInterrupted()} 判断当前线程是否被中断.
\end{enumerate}

\subsubsection{{\bfseries\sffamily TODO} \href{https://www.cnblogs.com/onlywujun/p/3565082.html}{参考资料}}
\label{sec:orgff19b5c}

\subsection{5.5. Synchronizers}
\label{sec:org458a856}

同步器是通过自身状态用来协调线程控制流的对象. 例如阻塞队列就是很好的一个同步器示例, 因为 \texttt{take} 和 \texttt{put} 方法会出于阻塞状态, 用以使得队列进入合适的状态(既不为空, 也会是满)

其他的同步器包括:

\begin{itemize}
\item 信号量(semaphores)

\item 屏障/栅栏??(barriers)

很多人都把Barrier直译为"栅栏"，我也很喜欢这个叫法。
\end{itemize}
栅栏可以使一组执行在一处汇集，也就是说我们可以用栅栏将一个问题分解成多个独立的子问题，并在执行结束后在同一处进行汇集。
当线程到达汇集地后调用await，await方法会出现阻塞直至其他线程也到达汇集地。
如果所有的线程都到达就可以通过栅栏，也就是所有的线程得到释放，而且栅栏也可以被重新利用。

A CyclicBarrier is a synchronizer that allows a set of threads to wait for each other to reach a common execution point, also called a barrier.)

\begin{itemize}
\item latches(门闩)

shuan——指门关上后，插在门内使门推不开的滑动插销

即，闭锁可以延迟线程执行直至达到相应的结束状态。闭锁就像一个大门，未到达结束状态相当于大门紧闭，不让任何线程通过。而到达结束状态后，大门敞开，让所有的线程通过，但是一旦敞开后不会再关闭。闭锁可以用来确保一些活动在某个事件发生后执行。
\end{itemize}


同步器虽然实现各不相同, 但是都遵循相似的原则: 他们通过状态来判断到达同步器的线程是否应该通过或等待；他们提供方法来操作状态；提供方法来等有效的等待同步器进入所需的状态.

\subsubsection{5.5.1. Latches(门闩)}
\label{sec:orgfc5ad54}

Latch是一次性用品, 一旦到达0就不可再恢复.

常用的 \texttt{CountDownLatch} 用来使得线程等待某个时间的发生, 当这个对象的值不为0时, 需要通过 \texttt{await()} 方法等待. 用于一些线程同时开始工作的场景, 或是线程间有相互依赖.

\begin{verbatim}
public class TestHarness {
  public long timeTasks(int nThreads, final Runnable task) throws InterruptedException {
    final CountDownLatch startGate = new CountDownLatch(1);
    final CountDownLatch endGate = new CountDownLatch(nThreads);

    for (int i = 0; i < n; i++) {
      Thread thread =
	  new Thread() {
	    @Override
	    public void run() {
	      try {
		startGate.await();
		try {
		  task.run();
		} finally {
		  endGate.countDown();
		}
	      } catch (InterruptedException e) {
		e.printStackTrace();
	      }
	    }
	  };
      thread.start();
    }
    long start = System.nanoTime();
    startGate.countDown();
    endGate.wait();
    long end = System.nanoTime();
    return end - start;
  }
}
\end{verbatim}

上面的实例中, 主线程设置了beginGate, 所有的子线程在beginGate降为0之前不能运行. 而子线程运行结束则会减小endGate, 当endGate降低为0时表示所有的线程都运行结束.

\subsubsection{5.5.2. FutureTask}
\label{sec:org7c1b245}

\texttt{FutureTask} 也可以想门闩一样工作. (\texttt{FutureTask} 实现了 \texttt{Future}, 描述抽象的计算结果). 由 \texttt{FutureTask} 表示的计算可以通过 \texttt{Callable} 来实现, 它有三个状态: 等待运行, 运行中, 完成. 完成这个词包含了所有完成的状态(???), 包括 普通完成, 取消, 异常.

\texttt{Future.get} 的行为依赖于任务的状态. 如果任务已经完成, \texttt{get} 立刻返回结果, 否则就会处于阻塞状态, 等待任务到达完成状态返回结果或是抛出异常. \texttt{FutureTask} 将执行线程的结果转移到接收结果的线程; \texttt{FutureTask} 的标准确保这个转移是安全的.

\texttt{FutureTask} 通过 \texttt{Executor} 框架来表示异步任务. 用来表示任意潜在的长时间计算, 在需要计算结果之前就先开始这些任务. 

Before calling launderThrowable, Preloader tests for the known checked exceptions and rethrows them. That leaves only unchecked exceptions, which Preloader handles by calling launderThrowable and throwing the result. If the Throwable passed to launderThrowable is an Error, launderThrowable rethrows it directly; if it is not a RuntimeException, it throws an IllegalStateException to indicate a logic error. That leaves only RuntimeException, which launderThrowable returns to its caller, and which the caller generally rethrows.

\begin{verbatim}
public class Preloader {
  private final FutureTask<ProductInfo> futureTask =
      new FutureTask<ProductInfo>(ProductInfo::loadProductInfo);
  private final Thread thread = new Thread(futureTask);

  public void start() {
    thread.start();
  }

  public ProductInfo get() throws InterruptedException {
    try {
      return futureTask.get();
    } catch (ExecutionException e) {
      Throwable course = e.getCause();
      throw launderThrowable(course);
    }
  }

  public static RuntimeException launderThrowable(Throwable t) {
    if (t instanceof RuntimeException) return (RuntimeException) t;
    else if (t instanceof Error) throw (Error) t;
    else throw new IllegalStateException("Not checked", t);
  }
}
\end{verbatim}

\subsubsection{5.5.3. Semaphores}
\label{sec:orgf608f7a}

\texttt{Counting semaphores} 用来控制可以访问某个资源的线程的数量, 或是在相同的时间采取行动. \texttt{Counting semaphores} 用来实现资源池或是强制保持集合的数量.

一个信号量管理一系列虚拟的许可, 最初的许可数量通过 \texttt{Semaphore} 的构造函数进行设置. 线程可以获得或归还权限. 如果没有许可可用, \texttt{acquire} 会阻塞直至获得可用的许可. \texttt{release} 方法归还许可给信号两. 退化的计数信号量称之为 binary semaphore, 它是计数器初始化为1的信号量. \texttt{binary semaphore} 用于非重入的互斥操作, 保持对资源的单一访问.


\begin{verbatim}
public class BoundedHashSet<T> {
  private final Set<T> set;
  private final Semaphore sem;

  public BoundedHashSet(int bound) {
    set = new HashSet<>();
    sem = new Semaphore(bound);
  }

  public boolean add(T o) throws InterruptedException {
    sem.acquire();
    boolean added = false;
    try {
      added = set.add(o);
      return added;
    } finally {
      if (!added) sem.release();
    }
  }

  public boolean remove(T o) {
    boolean removed = set.remove(o);
    if (removed) sem.release();
    return removed;
  }
}
\end{verbatim}

\subsubsection{5.5.4. Barriers}
\label{sec:orgfae2a41}

马具和门闩一样, 都是用来阻塞一组线程直到特定事件发生. 不同点在于 \texttt{Latches} 等待事件的发生而马具等待的是其他线程. 马具有点像在某个时间点指定地点会合: "所有人早上6点在麦当劳会合, 先到的人等着, 人到齐了我们再想做什么事情".

\texttt{CyclicBarrier} 允许固定数目的参与者反复在一个会合点进行会合, 可用于交互算法, 将一个问题分解成一定数目的独立小问题. 

注意下面代码中:

\begin{enumerate}
\item \texttt{this.barrier = new CyclicBarrier(count, mainBoard::commitNewValues);}

用来表示创建一个马具, 里面有 \texttt{count} 个干活的人, 等他们的活都干好后, 由最后一个线程执行 \texttt{mainBoard::commitNewValues} 方法.

\item \texttt{barrier.await();}

当前线程的工作已经完成, 那么表示到达了汇聚点. 所有线程都到达汇聚点后, 表示可以采取下一步动作: \texttt{mainBoard::commitNewValues}
\end{enumerate}

\begin{verbatim}
public class CellularAutomata {
  private final Board mainBoard;
  private final CyclicBarrier barrier;
  private final Worker[] workers;

  public CellularAutomata(Board board) {
    this.mainBoard = board;
    int count = Runtime.getRuntime().availableProcessors();
    this.barrier = new CyclicBarrier(count, mainBoard::commitNewValues);
    this.workers = new Worker[count];
    for (int i = 0; i < count; i++) workers[i] = new Worker(mainBoard.getSubBoard(count, i));
  }

  private class Worker implements Runnable {
    private final Board board;

    public Worker(Board board) {
      this.board = board;
    }

    public void run() {
      while (!board.hasConverged()) {
	for (int x = 0; x < board.getMaxX(); x++)
	  for (int y = 0; y < board.getMaxY(); y++) board.setNewValue(x, y, computeValue(x, y));
	try {
	  barrier.await();
	} catch (InterruptedException | BrokenBarrierException ex) {
	  return;
	}
      }
    }
  }

  public void start() {
    for (int i = 0; i < workers.length; i++) new Thread(workers[i]).start();
    mainBoard.waitForConvergence();
  }
}
\end{verbatim}

马具的另一个形式是 \texttt{Exchanger}, 双方在一定的时间点交换数据. 例如一个线程填充buffer, 一个线程消耗buffer, 填充线程在buffer满的时候进行交换, 消耗线程在buffer空的时候进行交换.

\subsection{5.6. Building an Efficient, Scalable Result Cache}
\label{sec:orgbe7ca83}

\subsubsection{Memoizer1 - synchronized whole calc}
\label{sec:org35860cd}

使用 \texttt{synchronized} 关键字为每次计算进行加锁, 导致每次请求都是顺序的, 并发性较差.

\url{./img/ch05fig02.gif}

\begin{verbatim}
public class Memoizer1<A, V> implements Computable<A, V> {
  private final Map<A, V> cache = new HashMap<>();
  private final Computable<A, V> c;

  public Memoizer1(Computable<A, V> c) {
    this.c = c;
  }

  @Override
  // 因为cache使用的是HashMap, 非线程安全, 所以加锁了
  public synchronized V compute(A args) throws InterruptedException {
    V result = cache.get(args);
    if (result == null) {
      result = c.compute(args);
      cache.put(args, result);
    }
    return result;
  }
}
\end{verbatim}

\subsubsection{Memoizer2}
\label{sec:orgf1d0047}

使用 \texttt{ConcurrentHashMap} 来提高并发性, 多个线程可以同时读取cache中的数据:

\url{./img/ch05fig03.gif}

\begin{verbatim}
public class Memoizer2<A, V> implements Computable<A, V> {
  private final Map<A, V> cache = new ConcurrentHashMap<>();
  private final Computable<A, V> c;

  public Memoizer2(Computable<A, V> c) {
    this.c = c;
  }

  @Override
  public V compute(A args) throws InterruptedException {
    V result = cache.get(args);
    if (result == null) {
      result = c.compute(args);
      cache.put(args, result);
    }
    return result;
  }
}
\end{verbatim}

\subsubsection{Memoizer3}
\label{sec:orgbebfdcb}

将计算(\texttt{FutureTask})放进cache中, 这样, 计算开始的时候, 其他线程可以感知得到, 就不用进行相同的计算, 只需要等待这个计算的结果就可以了. 只有当cache中没有这个计算时, 才开始计算.

\begin{verbatim}
public class Memoizer3<A, V> implements Computable<A, V> {
  private final Map<A, Future<V>> cache = new HashMap<>();
  private final Computable<A, V> c;

  public Memoizer3(Computable<A, V> c) {
    this.c = c;
  }

  @Override
  public V compute(A args) throws InterruptedException {
    Future<V> f = cache.get(args);
    if (f == null) {
      Callable<V> eval = () -> c.compute(args);
      FutureTask<V> ft = new FutureTask<>(eval);
      f = ft;
      cache.put(args, ft);
      ft.run();
    } else {
      System.out.println("find in cache.");
    }
    try {
      return f.get();
    } catch (ExecutionException e) {
      var cause = e.getCause();
      if (cause instanceof Error) throw (Error) cause;
      else if (cause instanceof RuntimeException) throw (RuntimeException) cause;
      else throw new IllegalStateException("unchecked", e);
    }
  }
}
\end{verbatim}

这个方案的小问题如下所示: 两个线程进行相同的计算, 时间差了一点点, 导致第二个线程没有感知到第一个线程在进行计算, 同时计算了两次:

\url{./img/ch05fig04.gif}

\subsubsection{Memoizer}
\label{sec:orgd69f6cb}

改进点:

\begin{enumerate}
\item \texttt{cache.putIfAbsent}

通过原子操作确定只有当不存在的时候再放入cache, 避免了上面的刷新操作. 如果已经在cache中存在这个key了, 那么只需要获取key对应的Future中的计算结果即可.
\end{enumerate}

\begin{verbatim}
public class Memoizer<A, V> implements Computable<A, V> {
  private final ConcurrentMap<A, Future<V>> cache = new ConcurrentHashMap<>();
  private final Computable<A, V> c;

  public Memoizer(Computable<A, V> c) {
    this.c = c;
  }

  @Override
  public V compute(A args) throws InterruptedException {
    while (true) {
      Future<V> f = cache.get(args);
      if (f == null) {
	Callable<V> eval =
	    () -> c.compute(args);
	FutureTask<V> ft = new FutureTask<>(eval);
	f = cache.putIfAbsent(args, ft);
	if (f == null) {
	  f = ft;
	  ft.run();
	}
	try {
	  return f.get();
	} catch (CancellationException e) {
	  cache.remove(args, f);
	} catch (ExecutionException e) {
	  var cause = e.getCause();
	  if (cause instanceof Error) throw (Error) cause;
	  else if (cause instanceof RuntimeException) throw (RuntimeException) cause;
	  else throw new IllegalStateException("Not checked", cause);
	}
      }
    }
  }
}
\end{verbatim}

\subsection{总结}
\label{sec:org7e63910}

\begin{itemize}
\item \emph{It's the mutable state, stupid.}

几乎所有的并发操作都会分解到协同方位可变状态, 可变状态越少, 越可确保线程安全.

\item \emph{Make fields final unless they need to be mutable.}

不可变的对象天然就是线程安全的. 所以尽量确保clas的成员变量都是 \texttt{final} 的.

\item \emph{Encapsulation makes it practical to manage the complexity.}

将数据封装在对象中, 更能控制不变式, 使得更好的同步状态变得简单.

\item \emph{Guard each mutable variable with a lock.}

\item \emph{Guard all variables in an invariant with the same lock.}

\item \emph{Hold locks for the duration of compound actions.}

\item \emph{A program that accesses a mutable variable from multiple threads without synchronization is a broken program.}

\item \emph{Don't rely on clever reasoning about why you don't need to synchronize.}

\item \emph{Include thread safety in the desing process -- or explicitly document that your class is not thread-safe}

\item \emph{Document your synchronization policy}
\end{itemize}

\section{Chapter 6. \emph{com.linchuan.concurrency.Task Execution}}
\label{sec:orgf6dfd83}

大多数的并发应用都围绕着任务的执行. 将一个大的应用细分成成小任务可以简化程序的组织, 便于错误定位及恢复, 为并行工作提供天然的结构.

\subsection{6.1. Executing Tasks in Threads}
\label{sec:orga2f4a64}

\subsubsection{6.1.1. 顺序执行任务}
\label{sec:orge48f639}

\begin{verbatim}
public class SingleThreadWebServer {
  public static void main(String[] args) throws IOException {
    ServerSocket socket = new ServerSocket(8011);
    while (true) {
      Socket connection = socket.accept();
      var input = connection.getInputStream();
      System.out.println(new String(input.readAllBytes()));
      //      System.out.println(input);
      connection.close();
    }
  }
}
\end{verbatim}

\subsubsection{6.1.2. 为任务显式创建线程}
\label{sec:org8a4bb40}

\begin{verbatim}
public class ThreadPerTaskWebServer {
  public static void handleRequest(Socket socket) {}

  public static void main(String[] args) throws IOException {
    ServerSocket socket = new ServerSocket(8081);
    while (true) {
      final Socket connection = socket.accept();
      Runnable task = () -> handleRequest(connection);
      new Thread(task).start();
    }
  }
}
\end{verbatim}

\subsubsection{6.1.3. 无限制创建线程的缺点}
\label{sec:orgfb7ea14}

\begin{enumerate}
\item 线程生命周期的负载.

线程创建和销毁是有代价的. 创建线程需要事件, 因此会导致处理响应的时延增加. 如果请求是频繁且轻量级的, 那么为每个请求创建一个新的线程会消耗可观的计算资源.

\item 消耗资源.

激活线程会消耗系统资源, 尤其是内存资源. 当 \texttt{runnable} 线程比可用处理器多的时候, 线程会处于闲置状态(无法得到调度). 处于闲置状态的线程过多, 会消耗大量内存, 为垃圾回收带来压力; 同时线程竞争CPU资源也会带来其他的资源开销.

\item 稳定性

线程的数目是有限制的, 这个限制由JVM参数, 栈大小, 以及操作系统等因素共同决定. 当创建的线程数超过这个限制, 会出发 \texttt{OutOfMemoryError} 错误.
\end{enumerate}


在某个点之前, 创建更多的线程可以提升吞吐量, 但是超过这个点, 则会导致应用不稳定, 崩溃, 内存不足等问题.

\subsection{6.2. The Executor Framework}
\label{sec:org24b0beb}

线程池提供线程管理的框架, 作为 \texttt{Executor} 框架的一部分, \texttt{java.util.concurrent} 提供了灵活的线程池实现, Java标准库的任务执行主要是使用 \texttt{Executor} 而非 \texttt{Thread}.

\begin{verbatim}
public interface Executor {
    void execute(Runnable command);
}
\end{verbatim}

\texttt{Executor} 提供了任务提交和任务执行的接偶方法, 实现了生命周期的支持, 以及为统计数据收集, 应用管理, 监控提供了钩子.

\texttt{Executor} 基于生产者-消费者模型, 提交任务的是生产者(生产需要完成的工作单元); 执行任务的线程是消费者 (消费工作单元). 

\subsubsection{6.2.1. Example: Web Server Using Executor}
\label{sec:org2ce96f4}

\begin{verbatim}
public class TaskExecutionWebServer {
  private static final int NTHREADS = 100; 
  private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);

  public static void main(String[] args) throws IOException {
    ServerSocket socket = new ServerSocket(8081);
    while(true) {
      final Socket connection = socket.accept();
      Runnable task = new Runnable() {
	@Override
	public void run() {
	  handleRequest(connection);
	}
      };
      exec.execute(task);
    }
  }
}
\end{verbatim}


有了 \texttt{Excutor} 接口后可以很方便的制定自己的执行策略:

如可以指定每请求新增线程:

\begin{verbatim}
public class ThreadPerTaskExecutor implements Executor {
    public void execute(Runnable r) {
	new Thread(r).start();
    }
}
\end{verbatim}

也可以指定顺序执行策略:

\begin{verbatim}
public WithinThreadExecutor implements Executor {
    public void execute(Runnble r) {
	r.run();
    }
}
\end{verbatim}

\subsubsection{6.2.2. Execution Policies}
\label{sec:orga786224}

将任务提交和任务执行解耦的价值在于: 这样使得可以更为方便的指定或是替换执行策略. 一个执行策略指明了任务执行的: \texttt{what, where, when, and how} 

\begin{enumerate}
\item 任务在什么线程中执行
\item 任务执行的顺序是什么: FIFO, LIFO, 优先级?
\item 多少个任务并行执行
\item 多少个任务可以被入队列被暂定
\item 如果一个任务被拒绝(因为系统超载), 那个任务需要被终止, 如何通知应用这种情况
\item 任务执行前完成后需要做什么动作
\end{enumerate}

\subsubsection{6.2.3. Thread Pools}
\label{sec:orgf4f8dc5}

线程池管理工作线程, 线程池绑定工作队列(work queue), 保存等待执行的线程. 工作线程的生命周期很简单: 请求工作队列中的下一个任务, 执行任务, 继续等待下一个任务.

可以通过调用 \texttt{Executors} 的以下静态工厂方法创建线程池:

\begin{itemize}
\item \texttt{newFixedThreadPool}

固定大小的线程池, 当任务提交时创建线程池, 到达最大线程数时保留所有线程(如果线程挂了, 添加新的线程池)

\item \texttt{newCachedThreadPool}

Cached线程池可以灵活的收割多余线程(池中的线程数超过处理需求), 也可以增加线程, 但是线程池的大小没有限制.

\item \texttt{newSingleThreadExecutor}

创建唯一个线程处理任务, 可以很好的保持同步. 任务根据工作队列的属性(FIFO, LIFO, proority order)进行处理.

\item \texttt{newScheduledThreadPool}

固定大小的线程池, 支持延时或周期性的任务处理(和Timer类似)
\end{itemize}

\subsubsection{6.2.4. Executor Lifecycle}
\label{sec:orgc3e50aa}

如何退出在执行的线程, 终止work Queue中的任务.

\texttt{ExecutorService} 接口扩展了 \texttt{Executor}, 增加了一系列的生命周期管理方法(包括一些任务提交的方法). 

\begin{verbatim}
public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
    // ... additional convenience methods for task submission
}
\end{verbatim}

\texttt{ExecutorService} 包含的生命周期有3个状态 - \emph{\texttt{running, shutting down, terminated}}:

\texttt{ExecutorService} 初始化为 \texttt{running} 状态. \texttt{shutdown} 方法会触发正常结束:试图取消任务, 并且不会开始排队中的任务.

在一个 \texttt{ExecutorService} 被shutdown后才提交的任务会被 \texttt{rejected execution handler} 处理, 它会discard任务或是导致 \texttt{execute} 方法抛出异常 \texttt{RejectedExecutionException}. 所有的任务都处理结束, \texttt{ExecutorService} 会迁移到 \texttt{terminated} 状态. 你可以通过 \texttt{awaitTermination} 方法等待 \texttt{ExecutorService} 进入到终止状态, 或是通过 \texttt{isTerminated} 方法查询其是否已经到了终止状态. 通常会在 \texttt{shutdown} 后立即调用 \texttt{awaitTermination} 方法来实现同步的 shuting down操作.

\begin{verbatim}
public class LifecycleWebServer {
  private final Logger logger = Logger.getLogger(this.getClass().getName());
  private final int NTHREADS = Runtime.getRuntime().availableProcessors() * 8;
  private final ExecutorService exec = Executors.newFixedThreadPool(NTHREADS);

  public void start() throws IOException {
    ServerSocket socket = new ServerSocket(8081);
    while (!exec.isShutdown()) {
      try {
	final Socket conn = socket.accept();
	exec.execute(() -> handleRequest(conn));
      } catch (RejectedExecutionException e) {
	if (!exec.isShutdown()) logger.log(Level.WARNING, "task submission rejected", e);
      }
    }
  }

  public void stop() {
    exec.shutdown();
  }
}
\end{verbatim}

\subsubsection{6.2.5. Delayed and Periodic Tasks}
\label{sec:orgfbfb6e3}

\subsection{6.3. Finding Exploitable parallelism}
\label{sec:org5651d63}
\end{document}
